package com.apps.sdses.flink141.sql;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author jiwei
 * @description
 * @date 2023/8/17 15:53
 */
public class FlinkFieldAddDemoSQL {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //主流
        tableEnv.executeSql("create temporary table node1(col1 String, col2 bigint, col3 timestamp) " +
                "with ('connector' = 'datagen', " +
                "'rows-per-second' = '10', " +
                "'fields.col1.length' = '5', " +
                "'fields.col2.min' = '101', " +
                "'fields.col2.max' = '109'" +
                ")");

        Table node1Table = tableEnv.from("node1");
        DataStream<Row> node1DS = tableEnv.toDataStream(node1Table);

        //cdc source
        Schema schema = Schema.newBuilder()
                .column("id", DataTypes.INT().notNull())
                .column("name", DataTypes.STRING())
                .column("description", DataTypes.STRING())
//                .column("ts", DataTypes.STRING())
                .primaryKey("id")
                .build();

        TableDescriptor descriptor = TableDescriptor.forConnector("mysql-cdc").schema(schema)
                .option("hostname", "localhost")
                .option("port", "3306")
                .option("username", "root")
                .option("password", "root")
                .option("database-name", "mydb")
                .option("table-name", "products")
                .build();
        tableEnv.createTemporaryTable("node2", descriptor);
//        Table node2Table = tableEnv.from(descriptor).where("id = 101");
//        DataStream<Row> node2DS = tableEnv.toChangelogStream(node2Table);

//        ValueState<Row> rowValueState = null;
//        node2DS.map(new RichMapFunction<Row, Object>() {
//            @Override
//            public void open(Configuration parameters) throws Exception {
//                ValueStateDescriptor<Row> rowState = new ValueStateDescriptor<>("rowState", BasicTypeInfo.of(Row.class));
//                getRuntimeContext().getState(rowState);
//            }
//            @Override
//            public Object map(Row value) throws Exception {
//                rowValueState.update(value);
//                return null;
//            }
//        });

//        Row row = rowValueState.value();
//        System.out.println(row);
//        node1DS.map(new RichMapFunction<Row, Row>() {
//            @Override
//            public Row map(Row value) throws Exception {
//                Row row = rowValueState.value();
//                System.out.println("===========row=======" + row);
//                return value;
//            }
//        }).print();






//        tableEnv.executeSql("CREATE TEMPORARY VIEW node3 as select node2.id, node2.name, 'v1', 'v2' from node2 where id in (101, 102)" );
//
//
////        tableEnv.executeSql("create temporary view node4 as select n ")
//
//
////        tableEnv.sqlQuery("select * from node1 full")
//
//        Table table = tableEnv.sqlQuery("select * from node3");
//        tableEnv.toChangelogStream(table).print();

        env.execute();

    }
}
